package cn.edu.gmall.realtime.app.func;

import cn.edu.gmall.realtime.bean.TableProcess;
import cn.edu.gmall.realtime.utils.MyHbaseUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.*;

public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {

    private final OutputTag<JSONObject> outputTag;
    private final MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    private static final HashSet hashSet = new HashSet();

    public TableProcessFunction(OutputTag<JSONObject> outputTag, MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.outputTag = outputTag;
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void processBroadcastElement(String jsonStr, Context context, Collector<JSONObject> collector) throws Exception {

        //将配置信息流中的数据转换为 JSON 对象 {"database":"","table":"","type","","data":{"":""}}
        JSONObject jsonObject = JSON.parseObject(jsonStr);

        //取出数据中的表名以及操作类型封装 key
        JSONObject data = jsonObject.getJSONObject("data");
        //取出 Value 数据封装为 TableProcess 对象
        TableProcess tableProcess = JSON.parseObject(data.toString(), TableProcess.class);

        String table = data.getString("source_table");
        String operation = data.getString("operate_type");
        String key = table + ":" + operation;
        BroadcastState<String, TableProcess> broadcastState = context.getBroadcastState(mapStateDescriptor);
        broadcastState.put(key, tableProcess);
        System.out.println(key + " " + tableProcess);
        if (TableProcess.SINK_TYPE_HBASE.equalsIgnoreCase(tableProcess.getSinkType())){
            MyHbaseUtil.makeTable(tableProcess.getSinkTable());
        }
    }


    @Override
    public void processElement(JSONObject jsonObject, ReadOnlyContext readOnlyContext, Collector<JSONObject> collector) throws Exception {
        //获取状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = readOnlyContext.getBroadcastState(mapStateDescriptor);
        //获取表名和操作类型
        String table = jsonObject.getString("table");
        String operation = jsonObject.getString("operation");
        String key = table + ":" + operation;
        //取出对应的配置信息数据
        TableProcess tableProcess = broadcastState.get(key);
        if (tableProcess != null) {
            //向数据中追加 sink_table 信息
            jsonObject.put("sink_table", tableProcess.getSinkTable());
            //根据配置信息中提供的字段做数据过滤
            filterColumn(jsonObject.getJSONObject("data"), tableProcess.getSinkColumns());
            //判断当前数据应该写往 HBASE 还是 Kafka
            if (TableProcess.SINK_TYPE_KAFKA.equals(tableProcess.getSinkType())) {
                //Kafka 数据,将数据输出到主流
                collector.collect(jsonObject);
            } else if (TableProcess.SINK_TYPE_HBASE.equals(tableProcess.getSinkType())) {
                //HBase 数据,将数据输出到侧输出流
                readOnlyContext.output(outputTag, jsonObject);
            }
        } else {
            if (!hashSet.contains("No Key " + key + " In Mysql!")){
                hashSet.add("No Key " + key + " In Mysql!");
                System.out.println("No Key " + key + " In Mysql!");
            }

        }
    }




    //根据配置信息中提供的字段做数据过滤
    private void filterColumn(JSONObject data, String sinkColumns) {
        //保留的数据字段
        String[] fields = sinkColumns.split(",");
        List<String> fieldList = Arrays.asList(fields);
        Set<Map.Entry<String, Object>> entries = data.entrySet();

        entries.removeIf(next -> !fieldList.contains(next.getKey()));
    }
}
